package edu.wisc.hevea.topic;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import java.util.concurrent.ConcurrentSkipListSet;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import edu.wisc.hevea.broker.MessageBroker;
import edu.wisc.hevea.exception.InvalidTopicException;
import edu.wisc.hevea.exception.ReplicationException;
import edu.wisc.hevea.hsql.Message;
import edu.wisc.hevea.messaging.MessagingServiceClient;


public class TopicManager implements Runnable {

	private static Log log = LogFactory.getLog(TopicManager.class);
	
	private String hostName;
	private Set<String> topics;
	private Map<String, TopicHandler> topicHandlerMap;
	private MessagingServiceClient messagingServiceClient;
	
	private ConcurrentSkipListSet<String> consumedMsgIds;

	public TopicManager(MessageBroker messageBroker) {
		this.hostName = messageBroker.getMyHostName();
		this.topics = new HashSet<String>();
		this.setConsumedMsgIds(new ConcurrentSkipListSet<String>());

		topics.add("TOPIC_1");

		topicHandlerMap = new HashMap<String, TopicHandler>();
		for (String topic : topics) {
			TopicHandler tH = new TopicHandler(topic, hostName);
			topicHandlerMap.put(topic, tH);
		}
		
		messagingServiceClient = messageBroker.getMessagingServiceClient();
	}

	public Message handleMessage(String topic, String content)
			throws InvalidTopicException, SQLException, ReplicationException {

		TopicHandler topicHandler = topicHandlerMap.get(topic);
		if (topicHandler == null) {
			log.error("No topic found!");
			throw new InvalidTopicException("No topic found!");
		}

		log.info("Putting msg to Pending!");
		Message m = topicHandler.putMsgInPending(content);
		log.info("Successfully put msg to Pending!" + m);
		
		if(!messagingServiceClient.replicateMessage(m.getMsgId(), topic, content)) {
			throw new ReplicationException("Unable to replicate!");
		}
		
		return m;
	}

	public void handleMessageReplication(Message m)
			throws InvalidTopicException, SQLException {

		TopicHandler topicHandler = topicHandlerMap.get(m.getTopic());
		if (topicHandler == null) {
			log.error("No topic found!");
			throw new InvalidTopicException("No topic found!");
		}

		log.info("Putting msg to Replication!");
		topicHandler.putMsgInReplicatedTable(m);
		log.info("Successfully put msg to Replication!" + m);
				
	}
	
	public void handleMessageConsumed(Message m)
			throws InvalidTopicException, SQLException {

		
		if(m.getMsgId().equals("-1")) {
			return;
		}
		
		TopicHandler topicHandler = topicHandlerMap.get(m.getTopic());
		if (topicHandler == null) {
			log.error("No topic found!");
			throw new InvalidTopicException("No topic found!");
		}

		if(!m.getMsgId().equals("-1")) {
			log.info("Putting msg to Consumed!");
			topicHandler.putMsgInConsumed(m);
			log.info("Successfully put msg to Consumed!" + m);
			
			if(!consumedMsgIds.contains(m.getMsgId())) {
				consumedMsgIds.add(m.getMsgId());	
			}					
		}
	}
	
	public void handleReplicatedMessagesConsumed(List<String> msgIds)
			throws InvalidTopicException, SQLException {
		
		if (topicHandlerMap.size() == 0) {
			log.error("No topic found!");
			throw new InvalidTopicException("No topic found!");
		}

		TopicHandler topicHandler = topicHandlerMap.entrySet().iterator()
				.next().getValue();
		log.info("Processing reconciled msgs!");
		topicHandler.putReconciledMsgs(msgIds);						
	}
	
	public Message handleGetMessage(String topic) throws InvalidTopicException {
		TopicHandler topicHandler = topicHandlerMap.get(topic);
		if (topicHandler == null) {
			log.error("No topic found!");
			throw new InvalidTopicException("No topic found!");
		}

		log.debug("Getting message!");
		return topicHandler.getMessageForConsumption();
	}
	
	private void reconcileConsumedMessagesWithOtherBrokers() {
		List<String> consumedMsgs = new ArrayList<String>();
		consumedMsgs.addAll(consumedMsgIds);
		consumedMsgIds.clear();
		messagingServiceClient.reconcileMessages(consumedMsgs);
	}
	
	public long getPendingQueueLength() {
		long queueLength = 0;
		for(TopicHandler tH : topicHandlerMap.values()) {
			queueLength += tH.getMsgQueue().size();
		}
		return queueLength;
	}

	@Override
	public void run() {		
		for(Entry<String,TopicHandler> entry : topicHandlerMap.entrySet()) {
			try {
				entry.getValue().putPendingMsgsInQueue();
				// log.info(entry.getValue().toString());
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		reconcileConsumedMessagesWithOtherBrokers();
	}

	public ConcurrentSkipListSet<String> getConsumedMsgIds() {
		return consumedMsgIds;
	}

	public void setConsumedMsgIds(ConcurrentSkipListSet<String> consumedMsgIds) {
		this.consumedMsgIds = consumedMsgIds;
	}
	
}
